package com.xxrl.shop.service;

import com.xxrl.shop.common.entity.ItemSimilarity;
import com.xxrl.shop.common.utils.SimilarityComputer;
import com.xxrl.shop.domain.ProductSimilar;
import com.xxrl.shop.domain.UserRating;
import com.xxrl.shop.repository.ProductSimilarRepository;
import com.xxrl.shop.repository.UserRatingRepository;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.mllib.recommendation.Rating;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * @author mis
 */
@Service
public class UserRatingService {
    private static final Logger logger = LoggerFactory.getLogger(UserRatingService.class);

    private final UserRatingRepository userRatingRepository;
    private final ProductSimilarRepository productSimilarRepository;
    private final RecommendService recommendService;
    private final SimilarityComputer similarityComputer;

    private String curFile;

    @Value("${my.spark.trainLimit}")
    private int trainLimit;

    public UserRatingService(UserRatingRepository userRatingRepository, ProductSimilarRepository productSimilarRepository, RecommendService recommendService) {
        this.userRatingRepository = userRatingRepository;
        this.productSimilarRepository = productSimilarRepository;
        this.recommendService = recommendService;
        similarityComputer = new SimilarityComputer();
        curFile = "cur_ratings";
    }

    /**
     * 当数据量达到一定条数时开始训练
     * 每星期定时检查一次
     */
    @Scheduled(cron = "0 * * * * 1")
    public void train() {
        String date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
        long count = userRatingRepository.count();
        logger.info("start training job at {}, count is {}", date, count);
        if (count < trainLimit) {
            logger.info("count not meet the limit, don't train");
            return;
        }
        // 0. create data
        List<UserRating> all = userRatingRepository.findAll();
        String newFile = DateFormatUtils.format(new Date(), "yyyy-MM-dd") + "_ratings";

        logger.info("start generate data from [{}] and database to [{}]", curFile, newFile);
        recommendService.generateNewData(newFile, curFile, all);
        curFile = newFile;
        // 1. train recommend model
        JavaRDD<Rating> ratingRdd = recommendService.trainModel(curFile);
        // 2. train product similarity model
        List<List<ItemSimilarity>> itemSimilarityLists = similarityComputer.train(ratingRdd, 5);
        // 3. delete old data
        clear();
        // 4. save new data into mongo
        List<ProductSimilar> productSimilarList = new ArrayList<>();
        for (List<ItemSimilarity> list : itemSimilarityLists) {
            if (list.isEmpty()) {
                continue;
            }
            int productId = list.get(0).getProductId1();
            productSimilarList.add(new ProductSimilar(list, productId));
        }
        productSimilarRepository.saveAll(productSimilarList);
    }

    public void saveRating(UserRating userRating) {
        userRatingRepository.save(userRating);
    }

    public void clear() {
        productSimilarRepository.deleteAll();
        userRatingRepository.deleteAll();
    }
}
